package com.iqiyi.pps.epg.core.service.program;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.iqiyi.pps.epg.api.model.web.program.TProgram;
import com.iqiyi.pps.epg.core.async.AsyncService;
import com.iqiyi.pps.epg.core.async.HeartBeatCenter;
import com.iqiyi.pps.epg.core.async.IAsyncTask;
import com.iqiyi.pps.epg.core.async.IHeartBeat;
import com.iqiyi.pps.epg.core.model.control.PlatformControl;
import com.iqiyi.pps.epg.core.model.control.ProgramPlatformControl;
import com.iqiyi.pps.epg.core.model.program.*;
import com.iqiyi.pps.epg.core.model.task.SimpleTask;
import com.iqiyi.pps.epg.core.model.task.SimpleTaskStatus;
import com.iqiyi.pps.epg.core.model.task.SimpleTaskType;
import com.iqiyi.pps.epg.core.rpc.qipu.QipuService;
import com.iqiyi.pps.epg.core.service.broadcast.BroadcastControlService;
import com.iqiyi.pps.epg.core.service.notifymail.NotifyMailService;
import com.iqiyi.pps.epg.core.service.task.SimpleTaskService;
import com.iqiyi.pps.epg.core.utils.*;
import com.iqiyi.pps.epg.core.utils.mail.MailConstant;
import org.hibernate.criterion.Criterion;
import org.hibernate.criterion.Order;
import org.hibernate.criterion.Restrictions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.sql.Timestamp;
import java.util.*;


/**
 * Created by karkarrotxu on 2015/12/15.
 */
@Lazy(false)
@Service
@Transactional
public class ProgramAsyncServiceImpl extends AsyncService implements InitializingBean, IHeartBeat, IProgramAsyncService {

    private static final Logger logger = LoggerFactory.getLogger(ProgramAsyncServiceImpl.class);
    private final static int HEARTBEAT_DELAY = 10 * 1000; //
    private final static int HEARTBEAT_DELAY2 = 240 * 1000; //
    private final static int HEARTBEAT_DELAY3 = 5 * 1000; //
    private long heartBeatTime = 0;
    private long heartBeatTime2 = 0;
    private long heartBeatTime3 = 0;
    private long heartBeatTime4 = 0;
    private long AT_ONCE_MOCKER_ID = -9999l;
    public static String ASSIGN_WORK_IP = "10.153.108.204";
    public static String ASSIGN_VV_WORK_IP = "10.153.132.18";
    public static String ASSIGN_DIRECTOR_WORK_IP = "10.153.132.19";

    @Autowired
    private SimpleTaskService simpleTaskService;

    @Autowired
    private QipuService qipuService;

    @Autowired
    private BroadcastControlService broadcastControlService;

    @Autowired
    private IProgramDirectorService programDirectorService;

    @Autowired
    private NotifyMailService notifyMailService;

    @Autowired
    private ProgramService programService;

    public void afterPropertiesSet() throws Exception {
        logger.info("ProgramAsyncServiceImpl afterPropertiesSet");
        HeartBeatCenter.add(this);
    }

    protected boolean queryEnable() {
        return getEnable("program");
    }

    @Override
    public void tick(long now) {
        if (now - heartBeatTime3 >= HEARTBEAT_DELAY) {
            heartBeatTime3 = now;
            logger.info("triggerVVOnce tick");
            triggerVVOnce();
        }
        if (now - heartBeatTime4 >= HEARTBEAT_DELAY) {
            heartBeatTime4 = now;
            logger.info("triggerDirectorOnce tick");
            triggerDirectorOnce();

        }
        if (now - heartBeatTime >= HEARTBEAT_DELAY) {
            heartBeatTime = now;
            logger.info("triggerOnce tick");
            triggerOnce();
        }
        if (now - heartBeatTime2 >= HEARTBEAT_DELAY2) {
            heartBeatTime2 = now;
            logger.info("triggerOverrideBroadcast RecycleStream tick");
            triggerOverrideBroadcast();
            triggerRecycleOnce();
        }
    }

    @Override
    public void triggerOnce() {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                executeSimpleTaskList(SimpleTaskType.PROGRAM_BROADCOAST_CONTROL.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    public void addTicketSysNotify(ProgramData programData,final long cQipuId,String operator) {
        final long programId = programData.getId();
        final String reporter = operator;
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                ProgramData notifyData = programService.findProgramDataById(programId);
                ApiTicketSysUtil.sendLiveNotify(notifyData, cQipuId, reporter);
                logger.info("addTicketSysNotify query");
            }
        });
    }
    public void addTicketSysCancelNotify(ProgramData programData) {
        final long programId = programData.getId();
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                ProgramData notifyData = programService.findProgramDataById(programId);
                ApiTicketSysUtil.sendLiveNotifyCancel(notifyData );
                logger.info("addTicketSysCancelNotify query");
            }
        });
    }


    @Override
    public void saveProgramLogList(final List<ProgramLog> programLogList) {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                long startTime = System.currentTimeMillis();
                logger.info("[saveProgramLogList] begin save :" + startTime);
                programService.saveProgramLogList(programLogList);
                logger.info("[saveProgramLogList] end cost :" + (System.currentTimeMillis() - startTime));
            }
        });
    }

    public void triggerVVOnce() {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                executeSimpleTaskList(SimpleTaskType.NOTIFY_VV.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    public void triggerDirectorOnce() {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                executeSimpleTaskList(SimpleTaskType.NOTIFY_DIRECTOR.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    public void triggerRecycleOnce() {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                executeSimpleTaskList(SimpleTaskType.NOTIFY_RECYCLE_STREAM_URL.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    public void triggerOverrideBroadcast() {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                executeSimpleTaskList(SimpleTaskType.CHANNEL_TO_PROGRAM.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), false);
            }
        });
    }

    public void triggerLogoUrlCheck() {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                executeSimpleTaskList(SimpleTaskType.CHECK_EPISODE_LOGO_URL.getTypeIndex(), SimpleTaskStatus.WAIT.getStatusIndex(), true);
            }
        });
    }

    @Override
    public void addTaskCheckProgramInLive(final long channelId, final boolean changeToRun, final String streamName) {
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                notifyProgramInLive(channelId, changeToRun, streamName);
            }
        });
    }

    @Override
    public void addAsyncById(long id) {
        final long startTime = System.currentTimeMillis();
        final long asyncId = id;
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                TProgram tProgram = programService.findTProgramById(asyncId, false);
                if (tProgram.getQipuId() == 0) {
                    return;
                }
                qipuService.syncProgramData(tProgram, false);
                logger.info("addSyncById cost time {}", System.currentTimeMillis() - startTime);
            }
        });
    }


    public void addVVNotify(ProgramData program) {
        SimpleTask simpleTask = new SimpleTask();
        simpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        simpleTask.setWorkIp(isProductionEnv() ? ASSIGN_VV_WORK_IP : simpleTask.getServerIp());
        simpleTask.setEntityId(program.getId());
        simpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        simpleTask.setTaskType(SimpleTaskType.NOTIFY_VV.getTypeIndex());
        simpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        simpleTask.setExpireTime(new Timestamp(System.currentTimeMillis() + DateTimeUtil.DAY_TIME));
        List<ProgramDataBase> programDataBaseList = programService.findDataBaseByProgramDataId(program.getId());
        JSONObject entity = ApiMqUtils.getVVObject(program, programDataBaseList);
        simpleTask.setParams(JSON.toJSONString(entity, SerializerFeature.WriteMapNullValue));
        simpleTaskService.saveTask(simpleTask);
        triggerVVOnce();
    }


    public void addLogoUrlCheck(ProgramData program) {
        SimpleTask simpleTask = new SimpleTask();
        simpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        simpleTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : simpleTask.getServerIp());
        simpleTask.setEntityId(program.getId());
        simpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        simpleTask.setTaskType(SimpleTaskType.CHECK_EPISODE_LOGO_URL.getTypeIndex());
        simpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        simpleTask.setExpireTime(new Timestamp(System.currentTimeMillis() + DateTimeUtil.DAY_TIME));
        simpleTask.setParams(program.getDefaultLogoUrl());
        simpleTaskService.saveTask(simpleTask);
        triggerLogoUrlCheck();
    }

    public void addDirectorNotify(ProgramData program) {
        SimpleTask simpleTask = new SimpleTask();
        simpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        simpleTask.setWorkIp(isProductionEnv() ? ASSIGN_DIRECTOR_WORK_IP : simpleTask.getServerIp());
        simpleTask.setEntityId(program.getId());
        simpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        simpleTask.setTaskType(SimpleTaskType.NOTIFY_DIRECTOR.getTypeIndex());
        simpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        simpleTask.setExpireTime(new Timestamp(System.currentTimeMillis() + DateTimeUtil.DAY_TIME));
        JSONObject entity = new JSONObject();
        entity.put("qipuId", program.getQipuId());
        entity.put("programName", program.getPname());
        entity.put("startTime", program.getStartPlayTime().getTime());
        entity.put("endTime", program.getStopPlayTime().getTime());
        simpleTask.setParams(JSON.toJSONString(entity));
        simpleTaskService.saveTask(simpleTask);
        triggerDirectorOnce();
    }

    @Override
    public void addRecycleStreamUrl(ProgramDirector programDirector) {
        if (programDirector.getRecycleUrl() == null || programDirector.getRecycleUrl().length() < 10) {
            logger.info("skip recycle stream url: id={}", programDirector.getId());
            return;
        }
        SimpleTask simpleTask = new SimpleTask();
        simpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        simpleTask.setWorkIp(isProductionEnv() ? ASSIGN_DIRECTOR_WORK_IP : simpleTask.getServerIp());
        simpleTask.setEntityId(programDirector.getId());
        simpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        simpleTask.setTaskType(SimpleTaskType.NOTIFY_RECYCLE_STREAM_URL.getTypeIndex());
        simpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        simpleTask.setExpireTime(new Timestamp(System.currentTimeMillis() + DateTimeUtil.DAY_TIME));
        JSONObject entity = new JSONObject();
        entity.put("qipuId", programDirector.getProgramQipuId());
        String originalUrl = programDirector.getRecycleUrl();
        entity.put("streamName", originalUrl.substring(originalUrl.lastIndexOf("/") + 1, originalUrl.indexOf("?")));
        entity.put("authCode", originalUrl.substring(originalUrl.indexOf("=") + 1));
        simpleTask.setParams(JSON.toJSONString(entity));
        simpleTaskService.saveTask(simpleTask);
        triggerRecycleOnce();
    }

    @Override
    public void addOverrideBroadcastControl(long channelId, List<PlatformControl> platformControls, boolean atOnce) {
        SimpleTask simpleTask = new SimpleTask();
        simpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        simpleTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : simpleTask.getServerIp());
        simpleTask.setEntityId(channelId);
        simpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        simpleTask.setTaskType(SimpleTaskType.CHANNEL_TO_PROGRAM.getTypeIndex());
        simpleTask.setParams(JSONArray.toJSONString(platformControls));
        simpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        simpleTask.setExpireTime(new Timestamp(System.currentTimeMillis() + DateTimeUtil.DAY_TIME));

        if (atOnce) {
            simpleTask.setId(AT_ONCE_MOCKER_ID);
            boolean flag = copyChannelBroadcast(simpleTask);
            logger.info("[ProgramAsyncServiceImpl][addOverrideBroadcastControl][atOnce][channelId={}][flag={}]", channelId, flag);
        } else {
            simpleTaskService.saveTask(simpleTask);
            triggerOverrideBroadcast();
        }

    }

    @Override
    public void addBroadCoastToQipu(ProgramData program, boolean needSync) {
        if (needSync) {
            broadcastControlService.writeQipuBroadcast(program);
        }
        SimpleTask simpleTask = new SimpleTask();
        simpleTask.setServerIp(RequestUtils.getLoadlIpByOS());
        simpleTask.setWorkIp(isProductionEnv() ? ASSIGN_WORK_IP : simpleTask.getServerIp());
        simpleTask.setEntityId(program.getId());
        simpleTask.setTaskStatus(SimpleTaskStatus.WAIT.getStatusIndex());
        simpleTask.setTaskType(SimpleTaskType.PROGRAM_BROADCOAST_CONTROL.getTypeIndex());
        simpleTask.setCreateTime(new Timestamp(System.currentTimeMillis()));
        simpleTask.setExpireTime(new Timestamp(System.currentTimeMillis() + DateTimeUtil.DAY_TIME));
        simpleTaskService.saveTask(simpleTask);
    }

    protected void executeSimpleTaskList(int type, int status, boolean needOverride) {
        logger.info("[executeSimpleTaskList] execute type:{},status={}", type, status);
        try {
            List<SimpleTask> resultList = simpleTaskService.findList(type, status);
            if (resultList == null || resultList.size() == 0) {
                logger.info("[result empty] resultList == null :" + (resultList == null));
                return;
            }
            logger.info("[executeSimpleTaskList] execute size:" + resultList.size());
            long currentTime = System.currentTimeMillis();
            if (needOverride) {
                Collections.reverse(resultList);
            }
            List<SimpleTask> updateList = new ArrayList<SimpleTask>(2048);
            Map<Long, SimpleTask> checkMap = new HashMap<Long, SimpleTask>(2048);
            List<SimpleTask> waitToDealList = new ArrayList<SimpleTask>(2048);
            for (SimpleTask simpleTask : resultList) {
                if (currentTime > simpleTask.getExpireTime().getTime()) {
                    simpleTask.setTaskStatus(SimpleTaskStatus.EXPIRED.getStatusIndex());
                    updateList.add(simpleTask);
                } else {
                    if (needOverride) {
                        if (!checkMap.containsKey(simpleTask.getEntityId())) {
                            checkMap.put(simpleTask.getEntityId(), simpleTask);
                        } else {
                            simpleTask.setTaskStatus(SimpleTaskStatus.OVERRIDE.getStatusIndex());
                            updateList.add(simpleTask);
                        }
                    } else {
                        waitToDealList.add(simpleTask);
                    }
                }
            }
            logger.info("[executeSimpleTaskList] overtime size:" + updateList);
            if (updateList.size() > 0) {
                simpleTaskService.batchUpdateList(updateList);
            }

            if (checkMap.size() > 0) {
                waitToDealList.addAll(checkMap.values());
            }

            if (waitToDealList.size() > 0) {
                updateList.clear();

                for (SimpleTask simpleTask : waitToDealList) {
                    if (executeSimpleTask(simpleTask)) {
                        updateList.add(simpleTask);
                    }
                }
                if (updateList.size() > 0) {
                    simpleTaskService.batchUpdateList(updateList);
                }
            }
        } catch (Exception e) {
            logger.error("[executeSimpleTaskList] error", e);
        }
        logger.info("[executeSimpleTaskList] execute end");
    }

    protected boolean setProgramBroadcast(long programId) {
        ProgramData program = programService.findProgramDataById(programId);
        boolean flag = broadcastControlService.writeQipuBroadcast(program);
        return flag;
    }

    protected void notifyProgramInLive(long channelId, boolean changeToRun, String streamName) {
        programService.notifyProgramInLive(channelId, changeToRun, streamName);
    }

    protected boolean copyChannelBroadcast(SimpleTask simpleTask) {
        logger.info("[copyChannelBroadcast][simpleTaskId={}][atOnce={}]", simpleTask.getId(), (AT_ONCE_MOCKER_ID == simpleTask.getId()));
        long startTime = System.currentTimeMillis();
        long maxTime = DateTimeUtil.getZeroTime(startTime) + DateTimeUtil.DAY_TIME * 2;
        List<Criterion> list = new ArrayList<Criterion>();
        list.add(Restrictions.eq("fatherId", (int) simpleTask.getEntityId()));
        list.add(Restrictions.between("startPlayTime", new Timestamp(startTime), new Timestamp(maxTime)));
        list.add(Restrictions.eq("deleteFlag", 0));
        list.add(Restrictions.gt("qipuId", 0L));
        List<Order> orders = new ArrayList<Order>();
        orders.add(Order.asc("startPlayTime"));
        List<ProgramData> programDatas = programService.queryDataByCriterion(list, orders, -1);
        List<ProgramPlatformControl> platformControls = JSONArray.parseArray(simpleTask.getParams(), ProgramPlatformControl.class);
        Map<Integer, ProgramPlatformControl> controlMap = new HashMap<>();
        for (ProgramPlatformControl control : platformControls) {
            control.setLiveDelayTime(0);
            controlMap.put(control.getPlayPlatform(), control);
        }
        logger.info("[copyChannelBroadcast] programDatas size :" + programDatas.size());
        long costStartTime = System.currentTimeMillis();
        boolean flag = true;
        Map<Integer, ProgramPlatformControl> copyMap = null;
        for (ProgramData programData : programDatas) {
            copyMap = resetIdZero(programData.getId(), controlMap);
            flag = broadcastControlService.setsWithProgram(programData.getId(), copyMap);
            if (!flag) {
                logger.info("[copyChannelBroadcast] break in programId:" + programData.getId());
                break;
            }
        }

        logger.info("[copyChannelBroadcast] programDatas size :{},cost:{}", programDatas.size(), (System.currentTimeMillis() - costStartTime));
        return flag;
    }

    protected boolean sendVVNotify(SimpleTask simpleTask) {
        long time1 = System.currentTimeMillis();
        ApiMqUtils.sendVVNotify(simpleTask.getParams());
        logger.info("[sendVVNotify]cost time {}", System.currentTimeMillis() - time1);
        return true;
    }

    protected boolean sendDirector(SimpleTask simpleTask) {
        long time1 = System.currentTimeMillis();
        JSONObject jsonObject = JSONObject.parseObject(simpleTask.getParams());
        ProgramDirector programDirector = programDirectorService.getByProgramQipuId(jsonObject.getLongValue("qipuId"));
        if (programDirector == null) {
            simpleTask.setTaskStatus(SimpleTaskStatus.ERROR.getStatusIndex());
            simpleTaskService.saveTask(simpleTask);
            return false;
        }
        ProgramDirectorLog logEntity = new ProgramDirectorLog();
        logEntity.setDirectorId(programDirector.getId());
        logEntity.setUser(Constants.SYSTEM_USER_SIGN);
        logEntity.setType(Constants.DirectorLogType.NOTIFY_EPISODE_CHANGE.getIndex());
        logEntity.setAddTime(new Timestamp(System.currentTimeMillis()));
        programDirector.setProgramName(jsonObject.getString("programName"));
        programDirector.setStartPlayTime(new Timestamp(jsonObject.getLongValue("startTime")));
        programDirector.setStopPlayTime(new Timestamp(jsonObject.getLongValue("endTime")));
        try {
            programDirectorService.updateProgramDirector(programDirector);
            logger.info("[sendDirector]cost time {}", System.currentTimeMillis() - time1);
            logEntity.setDescr("suc|| new|" + simpleTask.getParams() + "|| old |"
                    + programDirector.getProgramName() + "|" + programDirector.getStartPlayTime().getTime() + "|" + programDirector.getStopPlayTime().getTime());
            programDirectorService.addProgramDirectorLog(logEntity);
            return true;
        } catch (Exception e) {
            logger.error("[sendDirector]cost time:" + (System.currentTimeMillis() - time1), e);
            logEntity.setDescr("fail |" + simpleTask.getParams());
            programDirectorService.addProgramDirectorLog(logEntity);
            return false;
        }
    }

    protected boolean recycleStreamUrl(SimpleTask simpleTask) {
        long time1 = System.currentTimeMillis();
        JSONObject jsonObject = JSONObject.parseObject(simpleTask.getParams());
        boolean flag = false;
        try {
            flag = ApiDirectorUtil.unRegisterUrl(jsonObject.getString("streamName"), jsonObject.getString("authCode"));
            logger.info("[recycleStreamUrl][SUC]cost time {} [flag={}]", System.currentTimeMillis() - time1, flag);
        } catch (Exception e) {
            logger.error("[recycleStreamUrl][Fail]cost time:" + (System.currentTimeMillis() - time1), e);
        }
        return flag;
    }

    protected boolean checkEpisodeLogoUrl(SimpleTask simpleTask) {
        long time1 = System.currentTimeMillis();
        boolean flag = false;
        try {
            ProgramData data = programService.findProgramDataById(simpleTask.getEntityId());
            if (data.getDeleteFlag() == 1) {
                simpleTask.setTaskStatus(SimpleTaskStatus.INTERRUPT.getStatusIndex());
                simpleTaskService.saveTask(simpleTask);
            } else {
                if (simpleTask.getParams().equals(data.getDefaultLogoUrl())) {
                    flag = true;
                } else {
                    data.setDefaultLogoUrl(simpleTask.getParams());
                    programService.editProgramData(data);
                    logger.info("recall back fill {},{}", data.getId(), data.getDefaultLogoUrl());
                }
            }
            logger.info("cost time {} [flag={}]", System.currentTimeMillis() - time1, flag);
        } catch (Exception e) {
            logger.error("cost time:" + (System.currentTimeMillis() - time1), e);
        }
        return flag;
    }

    private Map<Integer, ProgramPlatformControl> resetIdZero(long programId, Map<Integer, ProgramPlatformControl> controlMap) {
        Map<Integer, ProgramPlatformControl> copyMap = new HashMap<>();
        ProgramPlatformControl copy = null;
        for (ProgramPlatformControl control : controlMap.values()) {
            copy = (ProgramPlatformControl) control.clone();
            copy.setId(0);
            copy.setProgramId(programId);
            copyMap.put(copy.getPlayPlatform(), copy);
        }
        return copyMap;
    }

    protected boolean executeSimpleTask(SimpleTask simpleTask) {
        SimpleTaskType type = SimpleTaskType.getTypeByIndex(simpleTask.getTaskType());
        boolean flag = false;
        switch (type) {
            case PROGRAM_BROADCOAST_CONTROL:
                flag = setProgramBroadcast(simpleTask.getEntityId());
                break;
            case CHANNEL_TO_PROGRAM:
                flag = copyChannelBroadcast(simpleTask);
                break;
            case NOTIFY_VV:
                flag = sendVVNotify(simpleTask);
                break;
            case NOTIFY_DIRECTOR:
                flag = sendDirector(simpleTask);
                break;
            case NOTIFY_RECYCLE_STREAM_URL:
                flag = recycleStreamUrl(simpleTask);
                break;
            case CHECK_EPISODE_LOGO_URL:
                flag = checkEpisodeLogoUrl(simpleTask);
                break;
        }
        if (flag) {
            simpleTask.setTaskStatus(SimpleTaskStatus.COMPLETE.getStatusIndex());
            simpleTask.setCompleteTime(new Timestamp(System.currentTimeMillis()));
        }
        return flag;
    }

    @Override
    public void addChangeTag(long programId) {
        final long newProgramId = programId;
        this.accept(new IAsyncTask() {
            @Override
            public void doTask() {
                setNewProgramTag(newProgramId);
            }
        });
    }

    private void setNewProgramTag(long programId) {
        logger.info("[setNewProgramTag][programId={}] START", programId);
        boolean flag = false;
        try {
            ProgramData programData = programService.findProgramDataById(programId);
            if (programData == null) {
                logger.info("programData is null .programId ={}", programId);
                return;
            }
            String oldDescr = programData.getProgramTag();
            String newDescr = "200003919_200978719";
            programData.setProgramTag(newDescr);
            sendNotifyTagChange(programData);
            programService.editProgramData(programData);
            flag = true;
            saveChangeTagLog(programId, newDescr, oldDescr);

        } catch (Exception e) {
            logger.error("exception={}", e);
        }
        logger.info("[setNewProgramTag][programId={}][flag={}] ", programId, flag);
    }

    private void sendNotifyTagChange(ProgramData programData) {
        Map<String, String> params = new HashMap<String, String>();
        params.put(MailConstant.P_ID, programData.getId() + "");
        params.put(MailConstant.C_ID, programData.getFatherId() + "");
        params.put(MailConstant.P_QIPU_ID, programData.getQipuId() + "");
        params.put(MailConstant.P_NAME, programData.getPname());
        params.put(MailConstant.ASSIGN_TITLE, String.format(MailContentUtil.getTitleTemplate(MailContentUtil.ContentType.CHANGE_TAG_REMIND), programData.getPname()));
        notifyMailService.addMailAsyncTask(MailConstant.TOS_TYPE_FROM_LOG, MailContentUtil.ContentType.CHANGE_TAG_REMIND.getTypeIndex(), params);
    }

    private void saveChangeTagLog(long programId, String newDescription, String oldDescription) {
        ProgramLog programLog = new ProgramLog();
        programLog.setProgramId(programId);
        programLog.setType(Constants.LogType.AUTO_SET_TAG.getIndex());
        programLog.setDescription("old tag:" + oldDescription + " || new tag: " + newDescription);
        programLog.setUser("_system_");
        programService.saveProgramLog(programLog);
    }

}
